package com.qiangesoft.rabbitmq.producer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

/**
 * 生产者
 *
 * @author qiangesoft
 * @date 2024-05-08
 */
@Slf4j
@RequestMapping("/producer")
@RestController
public class ProducerController {

    @Autowired
    public RabbitTemplate rabbitTemplate;

    @GetMapping("/send")
    public void send(String content) {
        CorrelationData correlation = getCorrelationData();

        Message message = MessageBuilder
                .withBody(content.getBytes(StandardCharsets.UTF_8))
                .setMessageId(UUID.randomUUID().toString())
                // 消息持久化
                .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                .build();

        // 正常发送
        rabbitTemplate.convertAndSend(MessageConfig.EXCHANGE, MessageConfig.ROUTING_KEY, message, correlation);
    }

    @GetMapping("/send1")
    public void send1(String content) {
        CorrelationData correlation = getCorrelationData();

        Message message = MessageBuilder
                .withBody(content.getBytes(StandardCharsets.UTF_8))
                .setMessageId(UUID.randomUUID().toString())
                // 消息持久化
                .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                .build();

        // 错误交换机发送
        rabbitTemplate.convertAndSend("fdff", MessageConfig.ROUTING_KEY, message, correlation);
    }

    @GetMapping("/send2")
    public void send2(String content) {
        CorrelationData correlation = getCorrelationData();

        Message message = MessageBuilder
                .withBody(content.getBytes(StandardCharsets.UTF_8))
                .setMessageId(UUID.randomUUID().toString())
                // 消息持久化
                .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                .build();

        // 错误路由key发送
        rabbitTemplate.convertAndSend(MessageConfig.EXCHANGE, "fsergrs", message, correlation);
    }

    private static CorrelationData getCorrelationData() {
        // 异步回调返回回执，开启publisher confirm机制【确保消息到达交换机】
        CorrelationData correlation = new CorrelationData();
        correlation.getFuture().addCallback(new ListenableFutureCallback<>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("消息发送异常，ID：{}，原因：{}", correlation.getId(), ex.getMessage());
            }

            @Override
            public void onSuccess(CorrelationData.Confirm result) {
                log.info("触发【publisher confirm】机制");
                if (result.isAck()) {
                    log.info("消息发送成功到达交换机，ID：{}", correlation.getId());
                } else {
                    // 消息发送失败
                    log.error("消息发送失败未到达交换机，ID：{}，原因：{}", correlation.getId(), result.getReason());
                }
            }
        });
        return correlation;
    }

}
